package edu.nepu.flink.api.windowjoin;

import edu.nepu.flink.api.bean.WaterSensor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SideOutputDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;

/**
 * @Date 2024/3/1 17:07
 * @Created by chenshuaijun
 */
public class IntervalJoinWindow {


    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> source1 = env.socketTextStream("hadoop102",8888).map(new MapFunction<String, WaterSensor>() {
            @Override
            public WaterSensor map(String value) throws Exception {
                String[] split = value.split(",");
                return new WaterSensor(split[0],Long.parseLong(split[1]),Integer.parseInt(split[2]));
            }
        });
        SingleOutputStreamOperator<WaterSensor> source2 = env.socketTextStream("hadoop102",9999).map(new MapFunction<String, WaterSensor>() {
            @Override
            public WaterSensor map(String value) throws Exception {
                String[] split = value.split(",");
                return new WaterSensor(split[0],Long.parseLong(split[1]),Integer.parseInt(split[2]));
            }
        });

        SingleOutputStreamOperator<WaterSensor> watermarksSource1 = source1.assignTimestampsAndWatermarks(WatermarkStrategy.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
            @Override
            public long extractTimestamp(WaterSensor element, long recordTimestamp) {
                return element.getTs() * 1000;
            }
        }));
        SingleOutputStreamOperator<WaterSensor> watermarksSource2 = source2.assignTimestampsAndWatermarks(WatermarkStrategy.<WaterSensor>forMonotonousTimestamps().withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
            @Override
            public long extractTimestamp(WaterSensor element, long recordTimestamp) {
                return element.getTs() * 1000;
            }
        }));

        /**
         * window join 的缺点是会对处于窗口边缘的数据不友好，导致join数据的缺失，同时拿不到join不上的数据
         * interval join可以对每一个数据划分一个区间，同时可以拿到join不上的数据
         * (1) interval join 只支持事件时间
         * (2) 需要指定上下界
         * (3) process中只能处理join上的数据
         * (4) 两条流join后watermark是以两条流中最小的为准
         * (5) 如果当前数据的事件事件 < 当前的watermark就是迟到的数据，会被放到侧输出流中。·
         */
        OutputTag<WaterSensor> leftTag = new OutputTag<WaterSensor>("left-miss-data"){};
        OutputTag<WaterSensor> rightTag = new OutputTag<WaterSensor>("right-miss-data"){};
        SingleOutputStreamOperator<String> process = watermarksSource1
                .keyBy(WaterSensor::getId)
                .intervalJoin(watermarksSource2.keyBy(WaterSensor::getId))
                .between(Time.seconds(-1), Time.seconds(2))
                .sideOutputLeftLateData(leftTag) // 使用侧输出流获取左侧没有join上的数据
                .sideOutputRightLateData(rightTag) // 使用侧输出流获取右侧没有join上的数据
                .process(new ProcessJoinFunction<WaterSensor, WaterSensor, String>() {
                    @Override
                    public void processElement(WaterSensor left, WaterSensor right, ProcessJoinFunction<WaterSensor, WaterSensor, String>.Context ctx, Collector<String> out) throws Exception {
                        out.collect("result :" + left + "-->" + right);
                    }
                });
        process.print("join上的结果：");
        SideOutputDataStream<WaterSensor> leftSideOutput = process.getSideOutput(leftTag);
        leftSideOutput.printToErr("左侧8888join不上的：");
        SideOutputDataStream<WaterSensor> rightSideOutput = process.getSideOutput(rightTag);
        rightSideOutput.printToErr("右侧9999join不上的：");


        env.execute();
    }
}
